package com.example.redislock.core;

import lombok.extern.slf4j.Slf4j;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.util.Assert;

import javax.annotation.PreDestroy;
import java.util.Collections;
import java.util.UUID;
import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * 基于Redis的可重入锁实现
 */
@Slf4j
public class RedisReentrantLock implements Lock {

    private static final String LOCK_PREFIX = "redis:lock:";
    private final ThreadLocal<LockInfo> threadLocal = new ThreadLocal<>();
    private static final ThreadLocal<String> THREAD_ID = ThreadLocal.withInitial(() -> 
        UUID.randomUUID() + ":" + Thread.currentThread().getId()
    );
    private final RedisTemplate<String, Object> redisTemplate;
    private final String baseKey;
    private final long defaultLockTimeout;
    private final long watchDogInterval;
    private final ScheduledExecutorService watchDogExecutor = Executors.newScheduledThreadPool(
            Runtime.getRuntime().availableProcessors(),
            new ThreadFactory() {
                private int counter = 0;
                @Override
                public Thread newThread(Runnable r) {
                    Thread thread = new Thread(r, "redis-lock-watchdog-" + counter++);
                    thread.setDaemon(true);
                    return thread;
                }
            }

    );
    
    private DefaultRedisScript<Long> lockScript;
    private DefaultRedisScript<Long> unlockScript;
    private DefaultRedisScript<Long> renewScript;

    public RedisReentrantLock(RedisTemplate<String, Object> redisTemplate, 
                             String lockKey, 
                             long defaultLockTimeout, 
                             long watchDogInterval) {
        Assert.notNull(redisTemplate, "RedisTemplate不能为空");
        Assert.hasText(lockKey, "锁键不能为空");
        Assert.isTrue(defaultLockTimeout > 0, "默认锁超时时间必须大于0");
        Assert.isTrue(watchDogInterval > 0 && watchDogInterval < defaultLockTimeout, 
                     "看门狗间隔必须大于0且小于锁超时时间");
        
        this.redisTemplate = redisTemplate;
        this.baseKey = LOCK_PREFIX + lockKey;
        this.defaultLockTimeout = defaultLockTimeout;
        this.watchDogInterval = watchDogInterval;
        log.debug("超时时间为"+String.valueOf(defaultLockTimeout));
        // 初始化Lua脚本
        initScripts();
    }
    
    /**
     * 初始化Lua脚本
     */
    private void initScripts() {
        try {
            lockScript = new DefaultRedisScript<>();
            lockScript.setLocation(new ClassPathResource("lua/lock.lua"));
            lockScript.setResultType(Long.class);

            unlockScript = new DefaultRedisScript<>();
            unlockScript.setLocation(new ClassPathResource("lua/unlock.lua"));
            unlockScript.setResultType(Long.class);

            renewScript = new DefaultRedisScript<>();
            renewScript.setLocation(new ClassPathResource("lua/renew.lua"));
            renewScript.setResultType(Long.class);

            log.info("Lua脚本加载完成");
        } catch (Exception e) {
            log.error("加载Lua脚本失败", e);
            throw new RedisLockException("加载Lua脚本失败", e);
        }
    }

    /**
     * 加锁（阻塞式）
     */
    @Override
    public void lock() {
        try {
            boolean locked = tryLock(defaultLockTimeout, TimeUnit.MILLISECONDS);
            if (!locked) {
                throw new RedisLockException("获取锁超时: " + baseKey);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RedisLockException("获取锁被中断: " + baseKey, e);
        }
    }

    /**
     * 尝试加锁（带超时）
     */
    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        long deadline = System.nanoTime() + unit.toNanos(time);
        String threadId = getThreadId();

        while (System.nanoTime() < deadline) {
            long remainingNanos = deadline - System.nanoTime();
            boolean success = tryAcquireLock(threadId, TimeUnit.NANOSECONDS.toMillis(remainingNanos));
            if (success) {
                LockInfo current = threadLocal.get();
                if (current == null) {
                    LockInfo newLock = new LockInfo(threadId);
                    threadLocal.set(newLock);
                    startWatchDog(newLock);
                    log.debug("首次获取锁成功");
                } else {
                    log.debug("重入锁成功");
                }
                return true;
            }
            if (remainingNanos <= 0) break;

            long sleepMs = Math.max(1L,
                    Math.min(10L,
                            TimeUnit.NANOSECONDS.toMillis(remainingNanos) / 2));
            TimeUnit.MILLISECONDS.sleep(sleepMs);
        }
        return false;
    }

    /**
     * 解锁
     */
    @Override
    public void unlock() {
        LockInfo currentLock = threadLocal.get();
        if (currentLock == null) {
            throw new RedisLockException("当前线程未持有锁: " + baseKey);
        }
        
        try {
            // 调用Redis释放锁，Redis会处理可重入计数
            Long result = tryReleaseLock(currentLock.getThreadId());
            
            if (result == null) {
                log.warn("释放锁时Lua脚本执行失败: {}", baseKey);
                throw new RedisLockException("释放锁失败: " + baseKey);
            }
            
            if (result == -1) {
                log.warn("锁不存在或不属于当前线程: {}", baseKey);
                throw new RedisLockException("锁不存在或不属于当前线程: " + baseKey);
            }
            
            if (result == 0) {
                // 锁完全释放，清理本地状态
                currentLock.stopWatchDog();
                threadLocal.remove();
                log.debug("锁完全释放: {}", baseKey);
            } else {
                // 重入锁释放，锁仍然存在，result是剩余计数
                log.debug("重入锁释放，剩余计数: {}, 锁键: {}", result, baseKey);
            }
            
        } catch (Exception e) {
            // 如果Redis操作失败，仍然清理本地状态
            currentLock.stopWatchDog();
            threadLocal.remove();
            throw new RedisLockException("释放锁时发生异常: " + baseKey, e);
        }
    }


    /**
     * 基于Redis加锁
     */
    private Boolean tryAcquireLock(String threadId, long timeoutMillis) {
        long timeoutSeconds = timeoutMillis / 1000;
        String timeoutStr = String.valueOf(timeoutSeconds);
        log.info("执行Lua脚本 - 锁键: {}, 线程ID: {}, 超时时间(秒): {}", baseKey, threadId, timeoutStr);

        Long result = redisTemplate.execute(
                lockScript,
                Collections.singletonList(baseKey),
                threadId,
                timeoutStr
        );

        log.info("Lua脚本执行结果: {}, 锁键: {}, 线程ID: {}", result, baseKey, threadId);

        if (result == null) {
            log.warn("尝试获取锁时Lua脚本执行失败，锁键: {}, 线程ID: {}", baseKey, threadId);
            return false;
        }

        if (result == -1) {
            log.error("Lua脚本参数错误，锁键: {}, 线程ID: {}, 超时时间(毫秒): {}, 超时时间(秒): {}", 
                    baseKey, threadId, timeoutMillis, timeoutSeconds);
            return false;
        }

        boolean success = result == 1;
        log.info("锁获取结果: {}, 锁键: {}, 线程ID: {}", success, baseKey, threadId);
        return success;
    }

    /**
     * 尝试释放锁
     * @return 返回值: -1=锁不存在或不属于当前线程, 0=锁完全释放, >0=重入锁释放但锁仍存在(返回剩余计数)
     */
    private Long tryReleaseLock(String threadId) {
        long timeoutSeconds = defaultLockTimeout / 1000;
        Long result = redisTemplate.execute(
                unlockScript,
                Collections.singletonList(baseKey),
                threadId,
                String.valueOf(timeoutSeconds)
        );
        
        if (result == null) {
            log.warn("尝试释放锁时Lua脚本执行失败，锁键: {}, 线程ID: {}", baseKey, threadId);
            return null;
        }
        
        log.debug("释放锁结果: {}, 锁键: {}, 线程ID: {}", result, baseKey, threadId);
        return result;
    }

    /**
     * 启动看门狗续期
     */
    private void startWatchDog(LockInfo lockInfo) {
        ScheduledFuture<?> future = watchDogExecutor.scheduleAtFixedRate(
                () -> {
                    try {
                        long timeoutSeconds = defaultLockTimeout / 1000;
                        Long result = redisTemplate.execute(
                                renewScript,
                                Collections.singletonList(baseKey),
                                lockInfo.getThreadId(),
                                String.valueOf(timeoutSeconds)
                        );
                        
                        if (result != null && result == 1) {
                            log.debug("锁续期成功: {}, 线程: {}", baseKey, lockInfo.getThreadId());
                        } else {
                            lockInfo.stopWatchDog();
                            log.warn("锁已不存在或不属于当前线程，停止续期: {}, 线程: {}", baseKey, lockInfo.getThreadId());
                        }
                    } catch (Exception e) {
                        log.error("锁续期失败: {}, 线程: {}, 错误: {}", baseKey, lockInfo.getThreadId(), e.getMessage());
                        lockInfo.stopWatchDog();
                    }
                },
                watchDogInterval,
                watchDogInterval,
                TimeUnit.MILLISECONDS
        );
        lockInfo.setWatchDogFuture(future);
    }

    /**
     * 获取线程唯一标识（JVM+线程ID）
     */
    private String getThreadId() {
        return THREAD_ID.get();
    }

    /**
     * 清理资源，关闭看门狗线程池
     */
    @PreDestroy
    public void destroy() {
        if (threadLocal.get() != null) {
            unlock();
        }
        log.info("开始清理Redis锁资源...");
        watchDogExecutor.shutdown();
        try {
            if (!watchDogExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
                log.warn("看门狗线程池未能在5秒内正常关闭，强制关闭");
                watchDogExecutor.shutdownNow();
            }
        } catch (InterruptedException e) {
            log.warn("等待看门狗线程池关闭时被中断，强制关闭");
            watchDogExecutor.shutdownNow();
            Thread.currentThread().interrupt();
        }
        log.info("Redis锁资源清理完成");
    }

    /**
     * 锁信息内部类 - 简化版本，只管理看门狗
     */
    private static class LockInfo {
        private final String threadId;
        private ScheduledFuture<?> watchDogFuture;

        public LockInfo(String threadId) {
            this.threadId = threadId;
        }

        public String getThreadId() {
            return threadId;
        }

        public void setWatchDogFuture(ScheduledFuture<?> future) {
            this.watchDogFuture = future;
        }

        public void stopWatchDog() {
            if (watchDogFuture != null && !watchDogFuture.isDone()) {
                watchDogFuture.cancel(true);
            }
        }
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {
        throw new UnsupportedOperationException("暂不支持可中断锁");
    }

    @Override
    public boolean tryLock() {
        String threadId = getThreadId();

        Boolean success = tryAcquireLock(threadId, defaultLockTimeout);
        if (Boolean.TRUE.equals(success)) {
            // 检查是否是首次获取锁
            LockInfo currentLock = threadLocal.get();
            if (currentLock == null) {
                // 首次获取锁，创建新的LockInfo并启动看门狗
                LockInfo newLock = new LockInfo(threadId);
                threadLocal.set(newLock);
                startWatchDog(newLock);
                log.info("首次获取锁成功!!! 锁键: {}, 线程ID: {}", baseKey, threadId);
            } else {
                // 重入锁，Redis已经处理了计数，这里只需要记录日志
                log.info("重入锁成功!!! 锁键: {}, 线程ID: {}", baseKey, threadId);
            }
            return true;
        }
        return false;
    }

    @Override
    public Condition newCondition() {
        throw new UnsupportedOperationException("暂不支持条件变量");
    }
}
